package drds.data_migrate.task_group;


import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.job.TaskGroup;
import drds.data_migrate.vo.State;
import drds.data_migrate.vo.Status;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.*;

@Author(name = Authors.LI_YANG)
@Slf4j
public class TaskGroupContainer
{


    /**
     * 当前taskGroup所属jobId
     */
    @Setter
    @Getter
    public long jobId;

    /**
     * 当前taskGroupId
     */
    @Setter
    @Getter
    public int taskGroupId;

    @Setter
    @Getter
    public TaskGroup taskGroup;

    private TaskMonitor taskMonitor = TaskMonitor.getInstance();

    public TaskGroupContainer(TaskGroup taskGroup)
    {
        super();


        this.jobId = 0;
        this.taskGroupId = 0;
        this.taskGroup = taskGroup;


    }


    public void run()
    {
        try
        {
            /**
             * 状态check时间间隔，较短，可以把任务及时分发到对应channel中
             */
            int sleepIntervalInMillSec = 100;
            /**
             * 状态汇报时间间隔，稍长，避免大量汇报
             */
            long reportIntervalInMillSec = 10000;
            /**
             * 2分钟汇报一次性能统计
             */

            // 获取channel数目
            int readWriteTaskNumberOfThisTaskGroup = 0;

            int taskMaxRetryTimes = 1;

            long taskRetryIntervalInMsec = 10000;

            long taskMaxWaitInMsec = 60000;

            List<TaskConfig> taskTaskConfigList = null;


            Map<Integer, TaskConfig> taskConfigMap = buildTaskConfigMap(taskTaskConfigList); //taskId与task配置
            List<TaskConfig> taskQueue = buildRemainTasks(taskTaskConfigList); //待运行task列表
            //
            Map<Integer, TaskExecutor> taskFailedExecutorMap = new HashMap<Integer, TaskExecutor>(); //taskId与上次失败实例
            List<TaskExecutor> taskExecutorList = new ArrayList<TaskExecutor>(readWriteTaskNumberOfThisTaskGroup); //正在运行task
            Map<Integer, Long> taskStartTimeMap = new HashMap<Integer, Long>(); //任务开始时间

            long lastReportTimeStamp = 0;
            State lastTaskGroupContainerState = new State();

            while (true)
            {
                //1.判断task状态
                boolean failedOrKilled = false;
                Map<Integer, State> taskIdToMessageMap = null;
                for (Map.Entry<Integer, State> entry : taskIdToMessageMap.entrySet())
                {
                    Integer taskId = entry.getKey();
                    State state = entry.getValue();
                    if (!state.isFinished())
                    {
                        continue;
                    }
                    TaskExecutor taskExecutor = removeTask(taskExecutorList, taskId);

                    //上面从runTasks里移除了，因此对应在monitor里移除
                    taskMonitor.removeTask(taskId);

                    //失败，看task是否支持failover，重试次数未超过最大限制
                    if (state.getStatus() == Status.failed)
                    {
                        taskFailedExecutorMap.put(taskId, taskExecutor);
                        if (taskExecutor.getWriteRunnable().supportFailOver() && taskExecutor.getAttemptCount() < taskMaxRetryTimes)
                        {
                            taskExecutor.shutdown(); //关闭老的executor
                            //containerCommunicator.resetMessage(taskId); //将task的状态重置
                            TaskConfig taskTaskConfig = taskConfigMap.get(taskId);
                            taskQueue.add(taskTaskConfig); //重新加入任务列表
                        } else
                        {
                            failedOrKilled = true;
                            break;
                        }
                    } else if (state.getStatus() == Status.killed)
                    {
                        failedOrKilled = true;
                        break;
                    } else if (state.getStatus() == Status.succeeded)
                    {
                        Long taskStartTime = taskStartTimeMap.get(taskId);
                        if (taskStartTime != null)
                        {
                            Long usedTime = System.currentTimeMillis() - taskStartTime;
                            log.info("taskGroup[{}] taskId[{}] is successed, used[{}]ms",
                                    this.taskGroupId, taskId, usedTime);
                            //usedTime*1000*1000 转换成PerfRecord记录的ns，这里主要是简单登记，进行最长任务的打印。因此增加特定静态方法
                            taskStartTimeMap.remove(taskId);
                            taskConfigMap.remove(taskId);
                        }
                    } else
                    {
                        throw new IllegalStateException();
                    }
                }

                // 2.发现该taskGroup下taskExecutor的总状态失败则汇报错误
                if (failedOrKilled)
                {


                    //throw DataExchangeException.asDataExchangeException(FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, lastTaskGroupContainerStateMessage.getThrowable());
                    throw new RuntimeException();
                }

                //3.有任务未执行，且正在运行的任务数小于最大通道限制
                Iterator<TaskConfig> iterator = taskQueue.iterator();
                while (iterator.hasNext() && taskExecutorList.size() < readWriteTaskNumberOfThisTaskGroup)
                {
                    TaskConfig taskTaskConfig = iterator.next();
                    Integer taskId = taskTaskConfig.getTaskId();
                    int attemptCount = 1;
                    TaskExecutor lastExecutor = taskFailedExecutorMap.get(taskId);
                    if (lastExecutor != null)
                    {
                        attemptCount = lastExecutor.getAttemptCount() + 1;
                        long now = System.currentTimeMillis();
                        long failedTime = lastExecutor.getState().getTimestamp();
                        if (now - failedTime < taskRetryIntervalInMsec)
                        {  //未到等待时间，继续留在队列
                            continue;
                        }
                        if (!lastExecutor.isShutdown())
                        { //上次失败的task仍未结束
                            if (now - failedTime > taskMaxWaitInMsec)
                            {
                                markMessageFailed();

                                // throw DataExchangeException.asDataExchangeException(CommonErrorCode.WAIT_TIME_EXCEED, "wrapper failover等待超时");
                                throw new RuntimeException();
                            } else
                            {
                                lastExecutor.shutdown(); //再次尝试关闭
                                continue;
                            }
                        } else
                        {
                            log.info("taskGroup[{}] taskId[{}] attemptCount[{}] has already shutdown",
                                    this.taskGroupId, taskId, lastExecutor.getAttemptCount());
                        }
                    }
                    TaskConfig taskTaskConfigForRun = taskMaxRetryTimes > 1 ? taskTaskConfig : taskTaskConfig;
                    TaskExecutor taskExecutor = new TaskExecutor(this, taskTaskConfigForRun, attemptCount);
                    taskStartTimeMap.put(taskId, System.currentTimeMillis());
                    taskExecutor.doStart();

                    iterator.remove();
                    taskExecutorList.add(taskExecutor);

                    //上面，增加task到runTasks列表，因此在monitor里注册。
                    taskMonitor.registerTask(taskId, null);

                    taskFailedExecutorMap.remove(taskId);
                    log.info("taskGroup[{}] taskId[{}] attemptCount[{}] is started",
                            this.taskGroupId, taskId, attemptCount);
                }

                //4.任务列表为空，executor已结束, 搜集状态为success--->成功
                if (taskQueue.isEmpty() && isAllTaskDone(taskExecutorList))
                {
                    // 成功的情况下，也需要汇报一次。否则在任务结束非常快的情况下，采集的信息将会不准确


                    log.info("taskGroup[{}] completed it's wrapper.", this.taskGroupId);
                    break;
                }

                // 5.如果当前时间已经超出汇报时间的interval，那么我们需要马上汇报
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > reportIntervalInMillSec)
                {


                    lastReportTimeStamp = now;

                    //taskMonitor对于正在运行的task，每reportIntervalInMillSec进行检查
                    for (TaskExecutor taskExecutor : taskExecutorList)
                    {
                        taskMonitor.report(taskExecutor.getTaskId(), null);
                    }

                }

                Thread.sleep(sleepIntervalInMillSec);
            }


        } catch (Throwable e)
        {
            State state = new State();

            if (state.getThrowable() == null)
            {
                state.setThrowable(e);
            }
            state.setState(Status.failed);


            throw new RuntimeException();
        } finally
        {
        }
    }

    private Map<Integer, TaskConfig> buildTaskConfigMap(List<TaskConfig> taskConfigList)
    {
        Map<Integer, TaskConfig> map = new HashMap<Integer, TaskConfig>();
        for (TaskConfig taskConfig : taskConfigList)
        {
            int taskId = taskConfig.getTaskId();
            map.put(taskId, taskConfig);
        }
        return map;
    }

    private List<TaskConfig> buildRemainTasks(List<TaskConfig> taskConfigList)
    {
        List<TaskConfig> remainTaskList = new LinkedList<TaskConfig>();
        for (TaskConfig taskConfig : taskConfigList)
        {
            remainTaskList.add(taskConfig);
        }
        return remainTaskList;
    }

    private TaskExecutor removeTask(List<TaskExecutor> taskExecutorList, int taskId)
    {
        Iterator<TaskExecutor> iterator = taskExecutorList.iterator();
        while (iterator.hasNext())
        {
            TaskExecutor taskExecutor = iterator.next();
            if (taskExecutor.getTaskId() == taskId)
            {
                iterator.remove();
                return taskExecutor;
            }
        }
        return null;
    }

    private boolean isAllTaskDone(List<TaskExecutor> taskExecutorList)
    {
        for (TaskExecutor taskExecutor : taskExecutorList)
        {
            if (!taskExecutor.isTaskFinished())
            {
                return false;
            }
        }
        return true;
    }


    private void markMessageFailed()
    {
        State state = new State();
        state.setState(Status.failed);
    }

}
